package com.beidu.lottery.call.service.portal.impl;

import com.alibaba.fastjson.JSONObject;
import com.beidu.lottery.call.constant.CacheKey;
import com.beidu.lottery.call.errorcode.CommonCode;
import com.beidu.lottery.call.constant.MQTopicsConst;
import com.beidu.lottery.call.dao.ActivityVoteGroupMapper;
import com.beidu.lottery.call.dao.ParticipationVoteMapper;
import com.beidu.lottery.call.dao.UserActivityMapper;
import com.beidu.lottery.call.exception.ServiceException;
import com.beidu.lottery.call.service.portal.ActivityVoteService;
import com.beidu.lottery.call.service.portal.UserDataService;
import com.beidu.lottery.call.service.portal.component.VoteComponent;
import com.beidu.lottery.call.utils.RedisUtil;
import com.beidu.lottery.call.utils.StockRedisUtil;
import com.github.pagehelper.PageInfo;
import com.beidu.lottery.call.utils.ActSequenceGen;
import com.beidu.lottery.call.utils.concurrent.RedissonClientUtil;
import com.beidu.lottery.call.entity.ParticipationVote;
import com.beidu.lottery.call.entity.PortalActivityRule;
import com.beidu.lottery.call.entity.dto.ActivityVoteDTO;
import com.beidu.lottery.call.entity.query.PortalGroupOptionQuery;
import com.beidu.lottery.call.entity.query.PortalOptionQuery;
import com.beidu.lottery.call.entity.vo.ActivityVoteGroupVO;
import com.beidu.lottery.call.entity.vo.PortalOptionVO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.List;

/**
 * <p>
 * 投票活动 服务实现类
 * </p>
 *
 * @author 北渡
 * @since 2023-08-23
 */
@Slf4j
@Service
public class ActivityVoteServiceImpl implements ActivityVoteService {

    @Resource
    private StockRedisUtil stockRedisUtil;

    @Resource
    private VoteComponent voteComponent;

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @Resource
    private ParticipationVoteMapper participationVoteMapper;

    @Resource
    private UserDataService userDataService;

    @Resource
    private ActivityVoteGroupMapper groupMapper;

    @Resource
    private UserActivityMapper userActivityMapper;

    @Resource
    private RedisUtil redisUtil;

    @Resource
    private RedissonClientUtil redissonClientUtil;

    @Override
    public String votePreSubmit(ActivityVoteDTO voteDTO, PortalActivityRule rule) {
        // 减扣投票次数
        if (!stockRedisUtil.decrLimit(CacheKey.genVoteCounter(voteDTO.getActivityId()), 1)) {
            throw new ServiceException(CommonCode.VOTE_TIMES_FULL.getMsg());
        }
        log.info("Redis中剩余投票次数-1");
        // 增加票数
        String key = CacheKey.genOptionsKey(voteDTO.getActivityId(), voteDTO.getGroupId());
        long row = voteComponent.increment(key, voteDTO.getOptionId(), 1);
        log.info("Redis中选项{}票数{}", voteDTO.getOptionId(), row);

        // 生成投票编号（MQ消息ID）
        String voteNumber = ActSequenceGen.genVoteNumber(voteDTO.getActivityId());
        log.info("生成投票编号, voteNumber => {}", voteNumber);
        voteDTO.setBaseDTO(voteNumber);// 设置用户信息上下文+MQ消息ID

        // 组装MQ消息
        JSONObject data = new JSONObject();
        data.put("dto", voteDTO);
        data.put("rule", rule);
        log.info("组装预投票MQ消息完成, data => {}", data);

        // 发送Kafka消息
        try {
            kafkaTemplate.send(MQTopicsConst.ACTIVITY_VOTE_SUBMIT, data.toJSONString());
        } catch (Exception e) {
            log.error("消息发送异常, 回滚投票次数, activityID => {}", voteDTO.getActivityId(), e);
            // 回滚投票次数
            stockRedisUtil.incr(CacheKey.genVoteCounter(voteDTO.getActivityId()), 1);
            // 回滚票数
            voteComponent.decrement(key, voteDTO.getOptionId(), 1);
            // 返回错误信息
            throw new ServiceException(CommonCode.RATE_LIMITER.getMsg());
        }

        log.info("预投票完成, result => {}", rule.getOperationTips());
        return rule.getOperationTips();
    }

    // seta分布式事务注解：@GlobalTransactional(name = "activity_vote_submit_logging", rollbackFor = Exception.class)
    @Transactional(rollbackFor = Exception.class)
    @Override
    public void voteSubmit(ActivityVoteDTO voteDTO, PortalActivityRule rule) {
        log.info("voteSubmit, voteDTO => {}", voteDTO);

        // 保存投票信息
        ParticipationVote vote = ParticipationVote.builder()
                .tenantCode(voteDTO.getTenantCode())
                .userId(voteDTO.getUserId())
                .activityId(voteDTO.getActivityId())
                .groupId(voteDTO.getGroupId())
                .optionId(voteDTO.getOptionId())
                .points(rule.getConsumePoints()).build();
        participationVoteMapper.insert(vote);
        log.info("保存投票信息完成, pId => {}", vote.getId());

        // 增加选项票数（投票计数）
//        int row = voteGroupOptionService.incrementVotes(voteDTO, 1);
        log.info("更新选项票数完成, row => {}", 1);

        // 调用API, 消费积分（负数）
        if (rule.getConsumePoints() != null && rule.getConsumePoints() != 0) {
            userDataService.pointsOpSubmit(-rule.getConsumePoints(), voteDTO.getActivityId());
        }
        log.info("投票入库完成, voteNumber => {}", voteDTO.getMessageId());
    }

    @Override
    public List<ActivityVoteGroupVO> queryVoteGroupCache(Integer id) {
        // 活动下架时清空缓存
        return redisUtil.computeIfAbsent(CacheKey.HOTSPOT + id + ":group", () -> groupMapper.selectListById(id));
    }

    @Override
    public PageInfo<PortalOptionVO> pageVoteOptionCache(PortalGroupOptionQuery param) {
        // 缓存Key: hotspot:{活动id}:{分组id}:options
        String key = CacheKey.genOptionsKey(param.getActivityId(), param.getGroupId());
        // 是否有缓存
        if (redisUtil.hasKey(key)) {
            return voteComponent.pageVoteOption(key, param);// 分页查询
        }
        // 分布式锁
        return redissonClientUtil.executeWithLock(key, () -> {
            // 双检
            if (redisUtil.hasKey(key)) {
                return voteComponent.pageVoteOption(key, param);
            }
            // 查询数据库中所有选项
            List<PortalOptionVO> list = userActivityMapper.pageVoteOption(param);
            if (list.isEmpty()) {
                return new PageInfo<>();
            }
            // 缓存所有选项
            voteComponent.cacheVoteOptions(key, list);
            log.info("缓存投票选项列表完成, size => {}", list.size());
            return voteComponent.pageVoteOption(key, param);
        });
    }

    @Override
    public PortalOptionVO queryVoteOptionCache(PortalOptionQuery param) {
        // 缓存Key: hotspot:{活动id}:{分组id}:options
        String key = CacheKey.genOptionsKey(param.getActivityId(), param.getGroupId());
        // 是否有缓存
        if (redisUtil.hasKey(key)) {
            return voteComponent.queryVoteOption(key, param);// 详情
        }
        // 分布式锁
        return redissonClientUtil.executeWithLock(key, () -> {
            // 双检
            if (redisUtil.hasKey(key)) {
                return voteComponent.queryVoteOption(key, param);
            }
            // 查询数据库中所有选项
            List<PortalOptionVO> list = userActivityMapper.pageVoteOption(
                    PortalGroupOptionQuery.builder().activityId(param.getActivityId()).groupId(param.getGroupId()).build()
            );
            if (list.isEmpty()) {
                return new PortalOptionVO();
            }
            // 缓存所有选项
            voteComponent.cacheVoteOptions(key, list);
            log.info("缓存投票选项列表完成, size => {}", list.size());
            return voteComponent.queryVoteOption(key, param);
        });
    }

}
